package com.javaspeak.concurrency.structures.arrayblockingqueue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.log4j.BasicConfigurator;

/**
 * @author John Dickerson
 */
public class ArrayBlockingQueuePerformanceTest {

    private Collection<WriteCallable> getWriteCallables(
         ArrayBlockingQueue<QueueObject> arrayBlockingQueue,
            int numberWritingThreads, int numberTradesPerWriteThread,
                TestScenario testScenario ){

        Collection<WriteCallable> writeCallables =
            new ArrayList<WriteCallable>( numberWritingThreads );

        List<QueueObject> queueObjects;
        QueueObject queueObject;

        for ( int i=0; i<numberWritingThreads; i++ ){

            queueObjects = new ArrayList<QueueObject>();

            for ( int j=0; j<numberTradesPerWriteThread; j++ ){

                queueObject = new QueueObject();
                queueObject.setTestScenario( testScenario );
                queueObjects.add( queueObject );
            }

            writeCallables.add(
                    new WriteCallable( arrayBlockingQueue, queueObjects ) );
        }

        return writeCallables;
    }


    private Collection<ReadCallable> getReadCallables(
            ArrayBlockingQueue<QueueObject> arrayBlockingQueue,
                int numberReadingThreads ){

        Collection<ReadCallable> readCallables =
            new ArrayList<ReadCallable>( numberReadingThreads );

        for ( int i=0; i<numberReadingThreads; i++ ){

            readCallables.add(
                  new ReadCallable( arrayBlockingQueue ) );
        }

        return readCallables;
    }


    private List<TestScenario> createTestScenarios(
            int numberQueueObjects, int maxNumberThreads, int numberSamples ){

        List<TestScenario> testScenarios = new ArrayList<TestScenario>();
        TestScenario testScenario;

        for ( int k=1; k<=maxNumberThreads; k++ ){

            for ( int l=1; l<=maxNumberThreads; l++ ){

                for ( int m=1; m<=numberSamples; m++ ){

                    testScenario  = new TestScenario();

                    testScenario.setNumberQueueObjects(
                            numberQueueObjects );

                    testScenario.setNumberWritingThreads( k );
                    testScenario.setNumberReadingThreads( l );

                    testScenarios.add( testScenario );
                }
            }
        }

        return testScenarios;
    }


    private void runTestScenario( TestScenario testScenario )
        throws InterruptedException {

        int numberQueueObjects = testScenario.getNumberQueueObjects();
        int numberWritingThreads = testScenario.getNumberWritingThreads();
        int numberReadingThreads = testScenario.getNumberReadingThreads();

        int numberTradesPerWriteThread =
            numberQueueObjects / numberWritingThreads;

        ExecutorService executorService =
            Executors.newFixedThreadPool(
                        numberWritingThreads + numberReadingThreads );

        ArrayBlockingQueue<QueueObject> arrayBlockingQueue =
            new ArrayBlockingQueue<QueueObject>( numberQueueObjects );

        Collection<WriteCallable> writeCallables =
            getWriteCallables( arrayBlockingQueue,
                    numberWritingThreads, numberTradesPerWriteThread,
                        testScenario );

        Collection<ReadCallable> readCallables =
            getReadCallables( arrayBlockingQueue, numberReadingThreads );

        List<Future<Result>> readResultFutures =
            new ArrayList<Future<Result>>();

        // start readCallable threads. These are the threads that read from
        // the queue. Note that the method submit does not wait for the threads
        // to end
        for ( ReadCallable readCallable : readCallables ){

            readResultFutures.add( executorService.submit( readCallable ) );
        }

        // now start the writeCallable threads.  Note that the invokeAll method
        // does wait for the threads to end or be interrupted
        @SuppressWarnings("unused")
        List<Future<Result>> writeResultFutures =
            executorService.invokeAll( writeCallables );

        // Now that the writeCallables have finished writing we need to tell
        // the readCallables to finish reading of the queue and then stop
        // when there are no more QueueObjects on it
        for ( ReadCallable readCallable : readCallables ){

            readCallable.keepRunning = false;
        }

        executorService.shutdownNow();

        Result readResult;

        long largestNanoProcessedTime = 0;
        long numberQueueObjectsRead = 0;

        // process results
        for ( Future<Result> readResultFuture : readResultFutures ){

            try {
                readResult = readResultFuture.get();

                if ( readResult.getNanoProcessedTime() > largestNanoProcessedTime ){

                    largestNanoProcessedTime = readResult.getNanoProcessedTime();
                }

                numberQueueObjectsRead =
                    numberQueueObjectsRead +
                        readResult.getNumberQueueObjectsProcessed();
            }
            catch( ExecutionException e ){

               throw new RuntimeException(
                       "unexpected exception reading trade" );
            }
        }

        System.out.println( "Processed "  +
                numberQueueObjectsRead + " queueObjects in " +
                    largestNanoProcessedTime / 1000000 +
                         " milliseconds : " +  testScenario.toString() );
    }


    public int factorial( int n ){

        if ( n == 0 ){

            return 1;
        }

        return n * factorial( n-1 );
    }


    public void runTestScenarios() throws InterruptedException {

        int maxNumberThreads = 6;
        int numberQueueObjects =  factorial( maxNumberThreads ) * 10000;
        int numberSamples = 2;

        List<TestScenario> testScenarios =
            createTestScenarios(
                    numberQueueObjects, maxNumberThreads, numberSamples );

        for ( TestScenario testScenario : testScenarios ){

            runTestScenario( testScenario );
        }
    }


    public static void main( String args[] ) throws InterruptedException {

        BasicConfigurator.configure();

        ArrayBlockingQueuePerformanceTest arrayBlockingQueuePerformanceTest =
            new ArrayBlockingQueuePerformanceTest();

        arrayBlockingQueuePerformanceTest.runTestScenarios();
    }
}
